package cool.taomu.toolkit.service.client.kademlia.utils

import com.google.common.base.Optional
import com.google.inject.Inject
import com.google.inject.name.Named
import cool.taomu.toolkit.retry.Retry
import cool.taomu.toolkit.rpc.thrift.service.ThriftConnectionFactory
import cool.taomu.toolkit.service.kademlia.entity.KPeerNode
import cool.taomu.toolkit.service.kademlia.service.PeerRateLimiter
import cool.taomu.toolkit.service.kademlia.thrift.KademliaService
import cool.taomu.toolkit.service.kademlia.thrift.Node
import cool.taomu.toolkit.service.kademlia.utils.NetwrokUtils
import cool.taomu.toolkit.utils.ByteBufferUtils
import java.net.SocketTimeoutException
import java.nio.charset.StandardCharsets
import java.util.concurrent.CompletableFuture
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit
import java.util.function.Function
import org.apache.commons.pool2.impl.GenericObjectPool
import org.apache.commons.pool2.impl.GenericObjectPoolConfig
import org.apache.thrift.protocol.TCompactProtocol
import org.apache.thrift.transport.TTransport
import org.apache.thrift.transport.layered.TFramedTransport
import org.eclipse.xtend.lib.annotations.Accessors
import org.slf4j.LoggerFactory
import cool.taomu.toolkit.service.kademlia.thrift.Response

@Accessors
class ServiceClientPool {

	val static LOG = LoggerFactory.getLogger(ServiceClientPool);

	@Inject
	@Named("taomu.kademlia.pool.config")
	GenericObjectPoolConfig<TTransport> config;

	@Inject
	ClientExec exec;

	val pool = new ConcurrentHashMap<String, GenericObjectPool<TTransport>>();

	def get(KPeerNode node, KPeerNode localNode) {
		LOG.info("获取一个Thrift服务的连接池对象")
		val pkey = #[node.host, node.port].join("#")
		var result = this.pool.computeIfAbsent(pkey, [
			LOG.info("无 {} 连接池,开始创建", pkey)
			if (!NetwrokUtils.isAvailable(node.host, node.port)) {
				LOG.info("尝试链接 {} 失败，尝试通过代理链接", node)
				NetwrokUtils.setProxyPeer(node)
			}
			if (NetwrokUtils.isAvailable(node.host, node.port)) {
				var factory = new ThriftConnectionFactory(node.host, node.port, 5000, [ socket |
					LOG.info("通过ping判断连接是否可用")
					try (var tTransport = new TFramedTransport(socket, Integer.MAX_VALUE)) {
						var protocol = new TCompactProtocol(tTransport);
						var client = new KademliaService.Client(protocol);
						var loaclNode = new Node(localNode.host, localNode.port)
						var resp = client.ping(loaclNode)
						var data = ByteBufferUtils.toString(resp.data, StandardCharsets.UTF_8)
						val pong = data.equals("PONG");
						if (pong) {
							LOG.info("ping {} 成功", pkey)
						}
						return pong
					} catch (Exception ex) {
						LOG.info("连接 {} 异常 {} \n判断是否在池中{}", pkey, ex.message, this.pool.containsKey(pkey));
						if (this.pool.containsKey(pkey)) {
							LOG.info("{} 连接出现异常,移除连接", pkey)
							this.pool.get(pkey).close
							this.pool.remove(pkey)
						} else {
							LOG.info("连接 {} 不在连接池中", pkey)
						}
						return false;
					}
				]);
				return factory.getObjectPool(config);
			}
		])
		return Optional.fromNullable(result)
	}

	val rateLimiter = new PeerRateLimiter(10);

	def synchronized <T> Optional<T> peerClient(KPeerNode node, KPeerNode localNode,
		Function<KademliaService.Client, T> func) {
		// 在这里加 Guava 限流
		if (!rateLimiter.getLimiter(node).tryAcquire(1, TimeUnit.SECONDS)) {
			LOG.warn("请求被限流，节点: {}", node);
			return Optional.absent();
		}
		var optional = this.get(node, localNode);
		if (optional.present) {
			var cpool = optional.get()
			var exec = exec.clientExec(cpool, func);
			val result = newArrayList();
			/*exec.whenComplete([ res, ex |
			 *     if (ex !== null) {
			 *         LOG.error(ex.message)
			 *     } else {
			 *         result.add(res)
			 *     }
			 ])*/
			result.add(exec.get(1, TimeUnit.MINUTES))
			LOG.info("result.size:{}", result.size)
			if (result.size > 0) {
				var er = result.get(0)
				/*if (er.get() instanceof Response) {
					LOG.info(
						"aaaa aaa result get(0):{}",
						ByteBufferUtils.toString((er.get() as Response).data, "UTF-8")
					)
				}*/
				return er
			}
		}
		return Optional.absent;
	}
}

class ClientExec {
	val static LOG = LoggerFactory.getLogger(ClientExec);

	@Retry(maxAttempts=3, delay=500, exponentialBackoff=true, rateLimit=2.0)
	def <T> CompletableFuture<Optional<T>> clientExec(GenericObjectPool<TTransport> cpool,
		Function<KademliaService.Client, T> func) {
		var socket = cpool.borrowObject
		try(var tTransport = new TFramedTransport(socket, Integer.MAX_VALUE)) {
			var protocol = new TCompactProtocol(tTransport);
			var client = new KademliaService.Client(protocol);
			try {
				var r = CompletableFuture.completedFuture(Optional.fromNullable(func.apply(client)));
				return r
			} catch (SocketTimeoutException stex) {
				throw stex
			}
		} catch (Exception ex) {
			if (socket !== null) {
				LOG.debug("socket标记为无效")
				cpool.invalidateObject(socket) // 标记为无效
			}
			throw ex
		} finally {
			if (socket !== null) {
				cpool.returnObject(socket)
			}
		}
	// return CompletableFuture.completedFuture(Optional.absent);
	}
}
